package com.examples.redis.reader;

import com.alibaba.fastjson.JSONObject;
import com.examples.mysql2mysql.stream.Flink2JdbcReader;
import com.examples.mysql2mysql.stream.MySink;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.util.Collector;

import java.util.Arrays;
import java.util.Set;
import java.util.function.Consumer;

public class RedisFlink {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream< String[]> dataStream = env
                .addSource(new Flink2RedisReader());
        dataStream.flatMap(new FlatMapFunction<String[], Object>() {
            @Override
            public void flatMap(String[] strings, Collector<
                    Object> collector) throws Exception {
                Arrays.asList(strings).forEach(new Consumer<String>() {
                    @Override
                    public void accept(String s) {
                        Object parse = JSONObject.parse(s);
                        if (parse instanceof String) {
                            parse = JSONObject.parse(String.valueOf(parse));
                        }
                        collector.collect(parse);
                    }
                });
            }
        }).addSink(new SinkFunction<Object>() {
            @Override
            public void invoke(Object value, Context context) throws Exception {
                SinkFunction.super.invoke(value, context);
                System.out.println("value = " + value);
            }
        });
        env.execute();
    }
}
